package com.ds.infrastructure.hyperspace.container.flow.handler.opentracing;

import com.ds.infrastructure.hyperspace.common.Container;
import com.ds.infrastructure.hyperspace.common.flow.FlowHandlerPipeline;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.handler.AbstractPreFlowHandler;
import com.ds.infrastructure.hyperspace.container.internal.opentracing.impl.TracerConstants;
import com.ds.infrastructure.hyperspace.container.internal.operator.OperatorDefinition;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.ASTERISK;
import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.EMPTY_STR;

/**
 * usage eg:
 * -   handler: opentracingMarkSampledPreHandler
 * handlerParams: [groupId1|groupId2|groupIdN, uri|headerName, (operatorName,operatorArgs)...]
 *
 * @author yaozh
 */
@Slf4j
public class OpentracingMarkSampledPreFlowHandler extends AbstractPreFlowHandler {
    private static final int INPUT_NAME_INDEX = 1;

    private Map<String, String> valueMap = new ConcurrentHashMap<>();

    private volatile String inputName;

    private volatile boolean allMatch;

    private volatile List<OperatorDefinition> operatorDefinitionList;

    @Override
    public void handleInput(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        if (duplexFlowContext.getRequestStartState() != data || duplexFlowContext.getTracerContext() == null) {
            return;
        }
        if (allMatch) {
            forceSampled(duplexFlowContext);
            return;
        }
        if (inputName != null) {
            String inputValue = extractAndComputeInput(duplexFlowContext, inputName, operatorDefinitionList);
            if (StringUtil.isNullOrEmpty(inputValue)) {
                return;
            }
            if (valueMap.containsKey(inputValue)) {
                forceSampled(duplexFlowContext);
            }
        }
    }

    private void forceSampled(HyperspaceDuplexFlowContext duplexFlowContext) {
        duplexFlowContext.getTracerContext().getGatewaySpan().setBaggageItem(
                TracerConstants.BAGGAGE_KEY_ZIPKIN_SAMPLED, TracerConstants.BAGGAGE_VALUE_SAMPLED);
    }

    @Override
    public String handlerName() {
        return "opentracingMarkSampledPreHandler";
    }

    @Override
    public void initHandlerParameters(Container container, FlowHandlerPipeline pipeline, String[] handlerParams) {
        String groupIdStr = handlerParams[0];
        if (ASTERISK.equals(groupIdStr)) {
            allMatch = true;
            return;
        }
        if (groupIdStr.indexOf(OR_DELIMITER) == -1) {
            valueMap.put(groupIdStr, EMPTY_STR);
        } else {
            String[] groupIdStrArray = StringUtils.delimitedListToStringArray(groupIdStr, OR_DELIMITER);
            for (String tempOneGroupIdStr : groupIdStrArray) {
                valueMap.put(tempOneGroupIdStr, EMPTY_STR);
            }
        }
        if (handlerParams.length > INPUT_NAME_INDEX) {
            inputName = handlerParams[INPUT_NAME_INDEX];
            operatorDefinitionList = parseOperatorDefinition(container, handlerParams, INPUT_NAME_INDEX + 1);
        }
    }
}
